package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

object Demo16MapPartitions {
  def main(args: Array[String]): Unit = {

    // 假设有一个黑名单：List[String]
    val blacklist: List[String] = List[String]("1500100001", "1500100011", "1500100111")

    /**
     * 基于黑名单结合SparkCore，从MySQL中获取学生的基本信息
     */

    // 除了使用textFile加载数据构建RDD，还可以通过Scala本地集合来构建RDD
    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo16MapPartitions")

    val sc: SparkContext = new SparkContext(conf)

    val blacklistRDD: RDD[String] = sc.parallelize(blacklist)

    //    blacklistRDD.foreach(println)

    // conn、pSt的构建不能放在map算子外面，不然就会导致Task未被序列化的问题
    // Task为什么需要被序列化？因为Task最终是需要由Driver端进行发送到Executor中执行，
    // Driver端和Executor之间交互是通过网络进行的，所以Task必须序列化后才能进行网络传输

    /**
     * 虽然Spark的代码都在一个main方法里，但实际上运行的时候会分为两个部分：
     * 算子内部的代码：最终会变成Task到Executor中执行
     * 算子外部的代码：在Driver端执行
     */

    blacklistRDD
      // 对每一条数据进行操作
      .map(id => {

        /**
         * 首先conn、pSt都是不可以放在算子外部的，因为它们不能够被序列化
         * 但是放算子里又会出现连接被重复创建的问题，相当于一条数据就需要建立一次连接
         */
        // 建立JDBC连接
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")

        // 创建prepareStatement
        val pSt: PreparedStatement = conn.prepareStatement("select name,age,gender,clazz from students where id = ?")

        pSt.setString(1, id)
        val rs: ResultSet = pSt.executeQuery()
        var returnStr: String = ""
        while (rs.next()) {
          val name: String = rs.getString("name")
          val age: Int = rs.getInt("age")
          val gender: String = rs.getString("gender")
          val clazz: String = rs.getString("clazz")
          returnStr = s"$id,$name,$age,$gender,$clazz"
        }
        returnStr
      })
    //      .foreach(println)

    // 统计班级里存在黑名单学生的人数

    // 使用mapPartitions算子进行优化：可以单独对每个分区的数据进行操作
    /**
     * 如果需要在Spark代码中同外部的某个数据源建立连接，
     * 连接建立之后如果是需要从外部获取数据一般就是使用mapPartitions
     * 如果是需要将结果数据写入外部系统，则一般使用foreachPartition
     */
    blacklistRDD
      .mapPartitions(iter => {

        // 建立JDBC连接
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")

        // 创建prepareStatement
        val pSt: PreparedStatement = conn.prepareStatement("select name,age,gender,clazz from students where id = ?")

        // iter表示RDD中一个分区的数据
        iter
          // 相当于对分区里的每一条数据进行操作，
          // 这里的map只是迭代器提供的一个普通方法而已，并不会有内部外部之分
          .map(id => {
            pSt.setString(1, id)
            val rs: ResultSet = pSt.executeQuery()
            var returnStr: String = ""
            while (rs.next()) {
              val name: String = rs.getString("name")
              val age: Int = rs.getInt("age")
              val gender: String = rs.getString("gender")
              val clazz: String = rs.getString("clazz")
              returnStr = s"$id,$name,$age,$gender,$clazz"
            }
            returnStr
          })
      })
      .foreach(println)

    blacklistRDD
      .repartition(2) // 对RDD进行重新分区
      .mapPartitionsWithIndex((index, iter) => {
        println(s"正在处理的分区index为：$index")
        iter
      }).foreach(println)


  }

}
